package com.spring.use.jms;

import com.spring.use.pojo.RocketMqMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * 消息生产者
 */
@Slf4j
@Component
public class MessageProducer {

    // 生产者 topic
    //private final String topic = "my-test-producer-topic";

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送不带tag的消息
     * @param msg
     * @param topic
     * @param group
     * @return: org.apache.rocketmq.client.producer.SendResult
     **/
    public <T> SendResult send(T msg, String topic, String group) {
        return this.send(msg, topic, group, null,false);
    }

    /**
     * 发送不带tag的事务消息
     * @param msg
     * @param topic
     * @param group
     * @return: org.apache.rocketmq.client.producer.SendResult
     **/
    public <T> SendResult send(T msg, String topic, String group,Boolean transaction) {
        return this.send(msg, topic, group, null,transaction);
    }

    /** topic 与 group 必须
     * 发送带tag的消息
     * @param msg
     * @param topic
     * @param group
     * @param tag
     * @param transaction 是否事务消息
     * @param <T>
     * @return
     */
    public <T> SendResult send(T msg, String topic, String group,String tag,Boolean transaction){
        if (StringUtils.isBlank(topic) || StringUtils.isBlank(group)){
            new Throwable("发送方topic或者group不能为空");
        }
        String uuid = UUID.randomUUID().toString().replaceAll("-", "");
        RocketMqMessage<T> mqMessage = new RocketMqMessage<>();
        mqMessage.setProducerGroup(group);
        mqMessage.setProducerTopic(topic);
        mqMessage.setProducerTag(tag);
        mqMessage.setContent(msg);
        mqMessage.setMsgKey(uuid);
        // 发送消息
        Message<RocketMqMessage<T>> messageFinal = MessageBuilder.withPayload(mqMessage).setHeader("KEYS", uuid).build();
        String destination = topic;
        if (StringUtils.isNotBlank(tag)){
            destination = topic +":" + tag;
        }
        //rocketMQTemplate.send(destination,messageFinal);
        if (!transaction){
            // 发送同步消息
            SendResult result = rocketMQTemplate.syncSend(destination, messageFinal);
            log.info("成功发送消息,发送响应MsgId:{},发送状态:{},消息内容为:{},返回值为:{}",result.getMsgId(),result.getSendStatus(), mqMessage, result);
            return result;
        }
        // 发送事务消息, 需要创建事务消息监听器
        TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(destination, messageFinal, tag);// 事物消息
        log.info("成功发送消息,发送响应MsgId:{},发送状态:{},消息内容为:{},返回值为:{}",transactionSendResult.getMsgId(),transactionSendResult.getSendStatus(), mqMessage, transactionSendResult);
        return transactionSendResult;
    }

    // =========================================== 异步消息 =============================================

    /**
     * 发送异步消息
     * @param msg
     * @param topic
     * @param group
     * @param sendCallback
     * @param <T>
     */
    public <T> void asyncSend(T msg, String topic, String group,SendCallback sendCallback){
        asyncSend(msg,topic,group,null,sendCallback);
    }

    /**
     * 发送异步消息
     * @param msg
     * @param topic
     * @param group
     * @param tag
     * @param sendCallback
     * @param <T>
     */
    public <T> void asyncSend(T msg, String topic, String group, String tag, SendCallback sendCallback){
        if (StringUtils.isBlank(topic) || StringUtils.isBlank(group)){
            new Throwable("发送方topic或者group不能为空");
        }
        String uuid = UUID.randomUUID().toString().replaceAll("-", "");
        RocketMqMessage<T> mqMessage = new RocketMqMessage<>();
        mqMessage.setProducerGroup(group);
        mqMessage.setProducerTopic(topic);
        mqMessage.setProducerTag(tag);
        mqMessage.setContent(msg);
        mqMessage.setMsgKey(uuid);
        // 发送消息
        Message<RocketMqMessage<T>> messageFinal = MessageBuilder.withPayload(mqMessage).setHeader("KEYS", uuid).build();
        String destination = topic;
        if (StringUtils.isNotBlank(tag)){
            destination = topic +":" + tag;
        }
        // 异步发送消息到，主线程不会被阻塞，立刻会返回
        rocketMQTemplate.asyncSend(destination, messageFinal,sendCallback);
        log.info("成功发送消息,发送消息内容为:{}", mqMessage);
    }
}
